/*

 * Licensed to the Apache Software Foundation (ASF) under one

 * or more contributor license agreements.  See the NOTICE file

 * distributed with this work for additional information

 * regarding copyright ownership.  The ASF licenses this file

 * to you under the Apache License, Version 2.0 (the

 * "License"); you may not use this file except in compliance

 * with the License.  You may obtain a copy of the License at

 *

 *     http://www.apache.org/licenses/LICENSE-2.0

 *

 * Unless required by applicable law or agreed to in writing, software

 * distributed under the License is distributed on an "AS IS" BASIS,

 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

 * See the License for the specific language governing permissions and

 * limitations under the License.

 */

package com.bff.gaia.unified.runners.gaia;



import com.bff.gaia.unified.model.pipeline.v1.RunnerApi;

import com.bff.gaia.unified.model.pipeline.v1.RunnerApi.Pipeline;

import com.bff.gaia.unified.runners.core.construction.graph.ExecutableStage;

import com.bff.gaia.unified.runners.core.construction.graph.GreedyPipelineFuser;

import com.bff.gaia.unified.runners.core.construction.graph.PipelineTrimmer;

import com.bff.gaia.unified.runners.fnexecution.jobsubmission.PortablePipelineRunner;

import com.bff.gaia.unified.runners.fnexecution.provisioning.JobInfo;

import com.bff.gaia.unified.sdk.PipelineResult;

import com.bff.gaia.unified.sdk.metrics.MetricsEnvironment;

import com.bff.gaia.api.common.JobExecutionResult;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;



import javax.annotation.Nullable;

import java.util.List;



import static com.bff.gaia.unified.runners.fnexecution.translation.PipelineTranslatorUtils.hasUnboundedPCollections;



/** Runs a Pipeline on Gaia via {@link GaiaRunner}. */

public class GaiaPipelineRunner implements PortablePipelineRunner {

  private static final Logger LOG = LoggerFactory.getLogger(GaiaPipelineRunner.class);



  private final GaiaPipelineOptions pipelineOptions;

  private final String confDir;

  private final List<String> filesToStage;



  public GaiaPipelineRunner(

	  GaiaPipelineOptions pipelineOptions, @Nullable String confDir, List<String> filesToStage) {

    this.pipelineOptions = pipelineOptions;

    this.confDir = confDir;

    this.filesToStage = filesToStage;

  }



  @Override

  public PipelineResult run(final Pipeline pipeline, JobInfo jobInfo) throws Exception {

    MetricsEnvironment.setMetricsSupported(false);



    GaiaPortablePipelineTranslator<?> translator;

    if (!pipelineOptions.isStreaming() && !hasUnboundedPCollections(pipeline)) {

      // TODO: Do we need to inspect for unbounded sources before fusing?

      translator = GaiaBatchPortablePipelineTranslator.createTranslator();

    } else {

      translator = new GaiaStreamingPortablePipelineTranslator();

    }

    return runPipelineWithTranslator(pipeline, jobInfo, translator);

  }



  private <T extends GaiaPortablePipelineTranslator.TranslationContext>

      PipelineResult runPipelineWithTranslator(

          final Pipeline pipeline, JobInfo jobInfo, GaiaPortablePipelineTranslator<T> translator)

          throws Exception {

    LOG.info("Translating pipeline to Gaia program.");



    // Don't let the fuser fuse any subcomponents of native transforms.

    Pipeline trimmedPipeline = PipelineTrimmer.trim(pipeline, translator.knownUrns());



    // Fused pipeline proto.

    // TODO: Consider supporting partially-fused graphs.

    RunnerApi.Pipeline fusedPipeline =

        trimmedPipeline.getComponents().getTransformsMap().values().stream()

                .anyMatch(proto -> ExecutableStage.URN.equals(proto.getSpec().getUrn()))

            ? trimmedPipeline

            : GreedyPipelineFuser.fuse(trimmedPipeline).toPipeline();



    GaiaPortablePipelineTranslator.Executor executor =

        translator.translate(

            translator.createTranslationContext(jobInfo, pipelineOptions, confDir, filesToStage),

            fusedPipeline);

    final JobExecutionResult result = executor.execute(pipelineOptions.getJobName());



    return GaiaRunner.createPipelineResult(result, pipelineOptions);

  }

}